package org.gorson.executor;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.java.io.TextInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;
import org.gorson.BO.OdsRepayPlanBO;
import org.gorson.function.FilterRepayPlanFunction;
import org.gorson.function.ProcessOdsReplyPlanFunction;

import java.io.File;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class FlinkSinkToEsDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        FileProcessingMode watchType = FileProcessingMode.PROCESS_ONCE;
        int interval = 10;
        final String matchLetter = ".*[a-wyzA-WYZ].*";
        final String matchLetter1 = "asset_side";

        String filePath = "D:\\flinkTest\\pl1.txt";
        SingleOutputStreamOperator<String> readFileSource = env
                .setParallelism(1)
                .readFile(new TextInputFormat((Path) null), filePath, watchType, interval)
                .name("source-custInfoBO").uid("source-custInfoBO");
//        readFileSource.print();

        SingleOutputStreamOperator<OdsRepayPlanBO> repayPlanSource = readFileSource.process(new ProcessOdsReplyPlanFunction())
                .filter(new FilterRepayPlanFunction());

        repayPlanSource.print();

        List<HttpHost> httpHosts = new ArrayList<>();
        httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));

/*        ElasticsearchSink.Builder<OdsRepayPlanBO> esSinkBuilder = new ElasticsearchSink.Builder<>(
                httpHosts,
                new ElasticsearchSinkFunction<OdsRepayPlanBO>() {

                    public IndexRequest createIndexRequest(OdsRepayPlanBO element) {
                        Map<String, String> json = new HashMap<>();
//                        json.put("data", element);
                        json.put("asset_side", element.getAssetSide());
                        json.put("credit_num", element.getCreditNum());
                        json.put("current_terms", element.getCurrentTerms());
                        json.put("pay_date", element.getDataDate());
                        json.put("pay_prin", element.getPayPrin());
                        json.put("pay_Inte", element.getPayInte());
                        json.put("pay_punit", element.getPayPunit());
                        json.put("pay_com_int", element.getPayComInt());
                        json.put("recv_pay_date", element.getRecvPayDate());
                        json.put("recv_pay_prin", element.getRecvPayPrin());
                        json.put("recv_pay_inte", element.getRecvPayInte());
                        json.put("recv_pay_punit", element.getRecvPayPunit());
                        json.put("recv_pay_com_int", element.getRecvPayComInt());
                        json.put("isclear", element.getIsclear());
                        json.put("ovd_flag", element.getOvdFlag());
                        json.put("data_date", element.getDataDate());
                        json.put("srart_date", element.getStartDate());

                        return Requests.indexRequest()
                                .index("repay_plan_20220113")
//                                .type("my-type")
                                .source(json, XContentType.JSON);
                    }

                    @Override
                    public void process(OdsRepayPlanBO odsRepayPlanBO, RuntimeContext runtimeContext, RequestIndexer indexer) {
                        indexer.add(createIndexRequest(odsRepayPlanBO));
                    }
                }
        );*/

        ElasticsearchSink.Builder<OdsRepayPlanBO> esSinkBuilder = new ElasticsearchSink.Builder<>(
                httpHosts,
                new ElasticsearchSinkFunction<OdsRepayPlanBO>() {

                    public IndexRequest createIndexRequest(String element) {
                        Map<String, String> json = new HashMap<>();
                        json.put("data", element);
/*                        json.put("asset_side", element.getAssetSide());
                        json.put("credit_num", element.getCreditNum());
                        json.put("current_terms", element.getCurrentTerms());
                        json.put("pay_date", element.getDataDate());
                        json.put("pay_prin", element.getPayPrin());
                        json.put("pay_Inte", element.getPayInte());
                        json.put("pay_punit", element.getPayPunit());
                        json.put("pay_com_int", element.getPayComInt());
                        json.put("recv_pay_date", element.getRecvPayDate());
                        json.put("recv_pay_prin", element.getRecvPayPrin());
                        json.put("recv_pay_inte", element.getRecvPayInte());
                        json.put("recv_pay_punit", element.getRecvPayPunit());
                        json.put("recv_pay_com_int", element.getRecvPayComInt());
                        json.put("isclear", element.getIsclear());
                        json.put("ovd_flag", element.getOvdFlag());
                        json.put("data_date", element.getDataDate());
                        json.put("srart_date", element.getStartDate());*/

                        return Requests.indexRequest()
                                .index("repay_plan_total_20220113")
//                                .type("my-type")
                                .source(json, XContentType.JSON);
                    }

                    @Override
                    public void process(OdsRepayPlanBO odsRepayPlanBO, RuntimeContext runtimeContext, RequestIndexer indexer) {
                        indexer.add(createIndexRequest(JSON.toJSONString(odsRepayPlanBO)));
                    }
                }
        );

        // configuration for the bulk requests; this instructs the sink to emit after every element, otherwise they would be buffered
        esSinkBuilder.setBulkFlushMaxActions(1);

// provide a RestClientFactory for custom configuration on the internally created REST client
//        esSinkBuilder.setRestClientFactory(
//                restClientBuilder -> {
//                    restClientBuilder.setDefaultHeaders("")
//                    restClientBuilder.setMaxRetryTimeoutMillis(...)
//                    restClientBuilder.setPathPrefix(...)
//                    restClientBuilder.setHttpClientConfigCallback(...)
//                }
//        );

// finally, build and add the sink to the job's pipeline
        repayPlanSource.addSink(esSinkBuilder.build());

        env.execute("");
    }

}
